package cc.verywell.pureblock.msg.nfs.map;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.msg.nfs.strc.EncodeAble;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV0;

/**
* Copyright © 2019 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2019年5月25日 下午4:45:35
*/
public class HDrelative implements FSrelative  {
	
	public static int DIRS_MAX = 1<<14; //Linux ext3 最大新建文件夹数为 31998
	public static int FILE_READ_BUFFER_SIZE = 50*1024;//文件缓存大小;
	
	public static String FILE_LINK_SUFFIX = ".link";
	public static String FILE_STATICS_SUFFIX = ".stat";
	public static String FILE_STATICS_TEMP_SUFFIX = ".stat.tmp";
	
	//public static int FILENAME_MAX = 128;
	//public static long FILE_CLUSTOR = 4*1024; //这个是磁盘一簇的空间， 一般系统是按此为单位占用空间的, 对于一些压缩文件夹，请调成1
	
	final protected Path filePath;
	private ReentrantReadWriteLock initLock;
	//CountDownLatch cdl = new CountDownLatch(3);
	
	private Map<String,RelativeCache> sizemap;
	static private class RelativeCache{
		public AtomicInteger v = new AtomicInteger(0);
		//TODO
	}

	public HDrelative(Path filePath) {
		this.filePath = filePath;
		this.initLock = new ReentrantReadWriteLock();
		Map<String,RelativeCache> map = new LinkedHashMap<>(DIRS_MAX>>4,0.9f,true);
		this.sizemap = Collections.synchronizedMap(map);
		
		this.initLock.writeLock().lock();
		initRun();/// 请在异步执行
	}
	
	public void initRun() {
		long time = Long.MAX_VALUE;
		try {
			time = System.currentTimeMillis(); //这个也是有可能出错的;
			Files.createDirectories(filePath);
		} catch (IOException e) {
			if(Pblog.FS)Pblog.FS(e);
		}finally {
			ReentrantReadWriteLock initLock = this.initLock;
			if(initLock!=null) {
				try {
					initLock.writeLock().unlock();
				}finally {
					this.initLock = null;
				}
			}
			if(Pblog.DEBUG)Pblog.DEBUG(this," init time: ",System.currentTimeMillis()-time);
		}
	}
	
	

	static public class MainSet implements RelativeDetail, EncodeAble{
		final public byte[] key;
		final public boolean isSet;
		final private long time;
		public MainSet(byte[] key,boolean isSet,long time) {
			this.key = key;
			this.isSet = isSet;
			this.time = time;
		};
		
		@Override
		public byte[] from() {return null;}
		@Override
		public byte state() {return isSet?RelativeDetail.STATE_ADDMAIN:RelativeDetail.STATE_CUTMAIN;}
		@Override
		public byte priority() {return 0;}
		@Override
		public long time() {return time;}
		
		/*
		 * (non-Javadoc)
		 * @see cc.verywell.pureblock.msg.nfs.strc.EncodeAble#encodeSize()
		 * 行长度4+HASHLEN2+HASH+STATE1+TIME8;
		 */
		@Override
		public int encodeSize() {
			return (4+2+1+8)+key.length;
		}
		@Override
		public void encode(ByteBuffer bb) {
			bb.putInt(encodeSize()-4);
			bb.putShort((short)key.length);
			bb.put(key);
			bb.put(state());
			bb.putLong(time);
		}
		
	}
	
	@Override
	public void putMainSet(byte[] to) {
		MainSet ms = new MainSet(to, true, System.currentTimeMillis());
		putMainSet(ms);
	}
	@Override
	public void removeMainSet(byte[] to) {
		MainSet ms = new MainSet(to, false, System.currentTimeMillis());
		putMainSet(ms);
	}
	public void putMainSet(MainSet ms) {
		ReentrantReadWriteLock initLock = this.initLock;
		if(initLock!=null) {
			try {
				initLock.readLock().lock();
			}finally {
				initLock.readLock().unlock();
			}
		}
		String dir = encodeDir(ms.key);
		Path d = filePath.resolve(dir+FILE_LINK_SUFFIX);
		RelativeCache fs = new RelativeCache();
		synchronized (sizemap) {
			RelativeCache fsold = sizemap.putIfAbsent(dir, fs);
			if(fsold!=null) {
				fs = fsold;
			}
			fs.v.getAndIncrement();
		}
		////在此期间 fs 可能已被删除
		synchronized (fs) { //在目录同步内进行目录操作;
			fs.v.getAndDecrement();
			if(Pblog.DEBUG)Pblog.DEBUG("hdrelative put main ",d);
			try {
				for(int i = 0;i<10;i++) {
					try {
						Files.createDirectories(filePath);
						if(!Files.exists(d)) {
							ByteBuffer bb = ByteBuffer.allocate(ms.encodeSize());
							ms.encode(bb);
							bb.flip();
							try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.CREATE,StandardOpenOption.WRITE);){
								Future<Integer> fu = afc.write(bb, 0);
								/*p = */fu.get();
							}
						}else {
							ByteBuffer bb = ByteBuffer.allocate((int)Math.min(FILE_READ_BUFFER_SIZE, Files.size(d)));
							long position = 0;
							try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.READ,StandardOpenOption.WRITE);){
								Future<Integer> fu = afc.read(bb, position);
								int r;
								boolean writeOk = ms.isSet;
								long readed = 0;
								rd:while((r=fu.get())>0) {
									position += r;
									bb.flip();
									while(bb.remaining()>4) {
										bb.mark();
										int msLen = bb.getInt();
										if(bb.remaining()>=msLen) {
											int next = bb.position()+msLen;
											toNext:{
												int l = 0xFF & bb.getShort();
												if(l!=ms.key.length)break toNext;
												for (byte b : ms.key) {
													if(b!=bb.get())break toNext;
												}
												switch(bb.get()) {
												case RelativeDetail.STATE_ADDMAIN:
													writeOk = false;
													break toNext;
												case RelativeDetail.STATE_CUTMAIN:
													writeOk = true;
													break toNext;
												}
											}
											bb.position(next);
											readed += 4+msLen;
										}else {
											if(msLen+4>FILE_READ_BUFFER_SIZE) {
												if(Pblog.FS)Pblog.FS("file structure err ",d," at ",readed," i will ignore rest.");
												break rd;
											}
											bb.reset();
											break;
										}
									}
									bb.compact();
									fu = afc.read(bb, position);
								}
								if(writeOk==ms.isSet) {
									position-=bb.remaining();
									bb.clear();
									ms.encode(bb);
									bb.flip();
									fu = afc.write(bb, readed);
									/*p = */fu.get();
									afc.truncate(readed+ms.encodeSize());
									afc.force(true);
								}
							}
						}
						return;
					}catch(IOException e1) {
						if(e1.getMessage().indexOf("many open file")<0) { //什么鬼，谁能解析一下？ 如何在系统排队?
							throw e1;
						}
					}
					Thread.sleep(i*100);
				}
			} catch (IOException e) {
				if(Pblog.FS)Pblog.FS(e);
			} catch (InterruptedException | ExecutionException e) {
				if(Pblog.DEBUG)Pblog.DEBUG(e);
			}
		}
	}
	
	static public class RelateSet implements RelativeDetail, EncodeAble{
		final public byte[] key;
		final private byte[] from;
		final public boolean isSet;
		final private byte priority;
		final private long time;
		public RelateSet(byte[] key, byte[] from, boolean isSet, byte priority, long time) {
			this.key = key;
			this.from = from;
			this.isSet = isSet;
			this.priority = priority;
			this.time = time;
		}
		
		@Override
		public byte[] from() {return from;}
		@Override
		public byte state() {return isSet?RelativeDetail.STATE_ADDLINK:RelativeDetail.STATE_CUTLINK;}
		@Override
		public byte priority() {return priority;}
		@Override
		public long time() {return time;}
		
		/*
		 * (non-Javadoc)
		 * @see cc.verywell.pureblock.msg.nfs.strc.EncodeAble#encodeSize()
		 * 行长度4+HASHLEN2+HASH+STATE1+PRIORITY1+FROMLEN2+FROM+TIME8;
		 */
		@Override
		public int encodeSize() {
			return (4+2+1+1+2+8)+key.length+from.length;
		}
		@Override
		public void encode(ByteBuffer bb) {
			bb.putInt(encodeSize()-4);
			bb.putShort((short)key.length);
			bb.put(key);
			bb.put(state());
			bb.put(priority);
			bb.putShort((short)from.length);
			bb.put(from);
			bb.putLong(time);
		}
	
	}
	@Override
	public void putRelative(byte[] from, byte[] to, byte priority) {
		RelateSet ms = new RelateSet(to, from, true, priority, System.currentTimeMillis());
		putRelative(ms);
	}
	@Override
	public void removeRelative(byte[] from, byte[] to, byte priority) {
		RelateSet ms = new RelateSet(to, from, false, priority, System.currentTimeMillis());
		putRelative(ms);
	}
	public void putRelative(RelateSet ms) {
		ReentrantReadWriteLock initLock = this.initLock;
		if(initLock!=null) {
			try {
				initLock.readLock().lock();
			}finally {
				initLock.readLock().unlock();
			}
		}
		String dir = encodeDir(ms.key);
		Path d = filePath.resolve(dir+FILE_LINK_SUFFIX);
		RelativeCache fs = new RelativeCache();
		synchronized (sizemap) {
			RelativeCache fsold = sizemap.putIfAbsent(dir, fs);
			if(fsold!=null) {
				fs = fsold;
			}
			fs.v.getAndIncrement();
		}
		////在此期间 fs 可能已被删除
		synchronized (fs) { //在目录同步内进行目录操作;
			fs.v.getAndDecrement();
			if(Pblog.DEBUG)Pblog.DEBUG("hdrelative put ",d);
			try {
				for(int i = 0;i<10;i++) {
					try {
						Files.createDirectories(filePath);
						if(!Files.exists(d)) {
							ByteBuffer bb = ByteBuffer.allocate(ms.encodeSize());
							ms.encode(bb);
							bb.flip();
							try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.CREATE,StandardOpenOption.WRITE);){
								Future<Integer> fu = afc.write(bb, 0);
								/*p = */fu.get();
							}
						}else {
							try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.READ,StandardOpenOption.WRITE);){
								ByteBuffer bb = ByteBuffer.allocate((int)Math.min(FILE_READ_BUFFER_SIZE, Files.size(d)));
								long position = 0;
								Future<Integer> fu = afc.read(bb, position);
								int r;
								boolean writeOk = ms.isSet;
								long readed = 0;
								rd:while((r=fu.get())>0) {
									position += r;
									bb.flip();
									while(bb.remaining()>4) {
										bb.mark();
										int msLen = bb.getInt();
										if(bb.remaining()>=msLen) {
											int next = bb.position()+msLen;
											toNext:{
												int l;
												l = 0xFF & bb.getShort();
												if(l!=ms.key.length)break toNext;
												for (byte b : ms.key) {
													if(b!=bb.get())break toNext;
												}
												boolean wrok = writeOk;
												switch(bb.get()) {
												case RelativeDetail.STATE_ADDLINK:
													wrok = false;
													break;
												case RelativeDetail.STATE_CUTLINK:
													wrok = true;
													break;
												default:
													break toNext;
												}
												if(bb.get()!=ms.priority())break toNext;
												l = 0xFF & bb.getShort();
												if(l!=ms.from.length)break toNext;
												for (byte b : ms.from) {
													if(b!=bb.get())break toNext;
												}
												writeOk = wrok;
											}
											bb.position(next);
											readed += 4+msLen;
										}else {
											if(msLen+4>FILE_READ_BUFFER_SIZE) {
												if(Pblog.FS)Pblog.FS("file structure err ",d," at ",readed," i will ignore rest.");
												break rd;
											}
											bb.reset();
											break;
										}
									}
									bb.compact();
									fu = afc.read(bb, position);
								}
								if(writeOk==ms.isSet) {
									position-=bb.remaining();
									bb.clear();
									ms.encode(bb);
									bb.flip();
									fu = afc.write(bb, readed);
									/*p = */fu.get();
									afc.truncate(readed+ms.encodeSize());
									afc.force(true);
								}
							}
						}
						return;
					}catch(IOException e1) {
						if(e1.getMessage().indexOf("many open file")<0) { //什么鬼，谁能解析一下？ 如何在系统排队?
							throw e1;
						}
					}
					Thread.sleep(i*100);
				}
			} catch (IOException e) {
				if(Pblog.FS)Pblog.FS(e);
			} catch (InterruptedException | ExecutionException e) {
				if(Pblog.DEBUG)Pblog.DEBUG(e);
			}
		}
	}
	
	static private class RelativeCountSet implements RelativeCount{
		public boolean isMainSet = false;
		public int directLink = 0;
		public int indirLink = 0;
		@Override
		public boolean isMainSet() {
			return isMainSet;
		}

		@Override
		public int directLink() {
			return directLink;
		}

		@Override
		public int indirLink() {
			return indirLink;
		}
	}
	@Override
	public RelativeCount getRelativeCount(byte[] key) {
		String dir = encodeDir(key);
		Path d = filePath.resolve(dir+FILE_LINK_SUFFIX);
		RelativeCountSet ms = new RelativeCountSet();
		if(!Files.exists(d))return ms;
		long position = 0;
		try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.READ);){
			ByteBuffer bb = ByteBuffer.allocate((int)Math.min(FILE_READ_BUFFER_SIZE, Files.size(d)));
			Future<Integer> fu = afc.read(bb, position);
			int r;
			long readed = 0;
			rd:while((r=fu.get())>0) {
				position += r;
				bb.flip();
				while(bb.remaining()>4) {
					bb.mark();
					int msLen = bb.getInt();
					if(bb.remaining()>=msLen) {
						int next = bb.position()+msLen;
						toNext:{
							int l;
							l = 0xFF & bb.getShort();
							if(l!=key.length)break toNext;
							for (byte b : key) {
								if(b!=bb.get())break toNext;
							}
							switch(bb.get()) {
							case RelativeDetail.STATE_ADDMAIN:
								ms.isMainSet = true;
								break toNext;
							case RelativeDetail.STATE_CUTMAIN:
								ms.isMainSet = false;
								break toNext;
							case RelativeDetail.STATE_ADDLINK:
								if(bb.get()==NfsStructureV0.DIRECTLINK_PRIORITY)
									ms.directLink++;
								else
									ms.indirLink++;
								break toNext;
							case RelativeDetail.STATE_CUTLINK:
								if(bb.get()==NfsStructureV0.DIRECTLINK_PRIORITY)
									ms.directLink--;
								else
									ms.indirLink--;
								break toNext;
							}
						}
						bb.position(next);
						readed += 4+msLen;
					}else {
						if(msLen+4>FILE_READ_BUFFER_SIZE) {
							if(Pblog.FS)Pblog.FS("file structure err ",d," at ",readed," i will ignore rest.");
							break rd;
						}
						bb.reset();
						break;
					}
				}
				bb.compact();
				fu = afc.read(bb, position);
			}
		}catch(IOException e) {
			if(Files.exists(d)) {
				if(Pblog.FS)Pblog.FS(e);
			}
		} catch (InterruptedException e) {
			if(Pblog.FS)Pblog.FS(e);
		} catch (ExecutionException e) {
			if(Pblog.FS)Pblog.FS(e);
		}
		return ms;
	}
	
	@Override
	public List<RelativeDetail> getRelativeList(byte[] key) {
		String dir = encodeDir(key);
		Path d = filePath.resolve(dir+FILE_LINK_SUFFIX);
		if(!Files.exists(d))return Collections.emptyList();
		long position = 0;
		ArrayList<RelativeDetail> al = new ArrayList<>();
		try(AsynchronousFileChannel afc = AsynchronousFileChannel.open(d,StandardOpenOption.READ);){
			ByteBuffer bb = ByteBuffer.allocate((int)Math.min(FILE_READ_BUFFER_SIZE, Files.size(d)));
			Future<Integer> fu = afc.read(bb, position);
			int r;
			long readed = 0;
			rd:while((r=fu.get())>0) {
				position += r;
				bb.flip();
				while(bb.remaining()>4) {
					bb.mark();
					int msLen = bb.getInt();
					if(bb.remaining()>=msLen) {
						int next = bb.position()+msLen;
						toNext:{
							int l;
							l = 0xFF & bb.getShort();
							if(l!=key.length)break toNext;
							for (byte b : key) {
								if(b!=bb.get())break toNext;
							}
							boolean isSet = true;
							switch(bb.get()) {
							case RelativeDetail.STATE_ADDMAIN:
								al.add(new MainSet(key, true, bb.getLong()));
								break toNext;
							case RelativeDetail.STATE_CUTMAIN:
								al.add(new MainSet(key, false, bb.getLong()));
								break toNext;
							case RelativeDetail.STATE_ADDLINK:
								//isSet = true;
								break;
							case RelativeDetail.STATE_CUTLINK:
								isSet = false;
								break;
							default:
								break toNext;
							}
							byte pri = bb.get();
							l = 0xFF & bb.getShort();
							byte[] from = new byte[l];
							bb.get(from);
							al.add(new RelateSet(key, from, isSet, pri, bb.getLong()));
						}
						bb.position(next);
						readed += 4+msLen;
					}else {
						if(msLen+4>FILE_READ_BUFFER_SIZE) {
							if(Pblog.FS)Pblog.FS("file structure err ",d," at ",readed," i will ignore rest.");
							break rd;
						}
						bb.reset();
						break;
					}
				}
				bb.compact();
				fu = afc.read(bb, position);
			}
		}catch(IOException e) {
			if(Files.exists(d)) {
				if(Pblog.FS)Pblog.FS(e);
			}
		} catch (InterruptedException e) {
			if(Pblog.FS)Pblog.FS(e);
		} catch (ExecutionException e) {
			if(Pblog.FS)Pblog.FS(e);
		}
		return al;
	}



	////////////////////////////////////////////////////////////////////
	/**
	 * 可重载，如果文件名太短，或许需要数据库作二级支撑
	 */
	public String encodeDir(byte[] key) {
		int hash = Arrays.hashCode(key) % DIRS_MAX;
		if(hash<0)hash=~hash;
		return Integer.toHexString(hash);
	}
	
	
	@Override
	public String toString() {
		StringBuffer sb = new StringBuffer();
		synchronized(sizemap) {
			sb.append("HDrelative dirs:").append(sizemap.size()).append("/").append(DIRS_MAX);
			//sizemap.keySet().stream().forEach(a->{sb.append(" ").append(a);});
		}
		return sb.toString();
	}
	
	
 }
